package server

import (
	"context"
	"github.com/pingcap-incubator/tinykv/kv/storage"
	"github.com/pingcap-incubator/tinykv/proto/pkg/kvrpcpb"
)

// The functions below are Server's Raw API. (implements TinyKvServer).
// Some helper methods can be found in sever.go in the current directory

// RawGet return the corresponding Get response based on RawGetRequest's CF and Key fields
func (server *Server) RawGet(_ context.Context, req *kvrpcpb.RawGetRequest) (*kvrpcpb.RawGetResponse, error) {
	// Your Code Here (1).
	//server.storage.Reader()
	reader, err := server.storage.Reader(nil)
	if err != nil {
		return nil, err
	}
	defer reader.Close()
	val, _ := reader.GetCF(req.GetCf(), req.GetKey())
	reps := new(kvrpcpb.RawGetResponse)
	if val == nil {
		reps.NotFound = true
	} else {
		reps.NotFound = false
		reps.Value = val
	}
	return reps, nil
}

// RawPut puts the target data into storage and returns the corresponding response
func (server *Server) RawPut(_ context.Context, req *kvrpcpb.RawPutRequest) (*kvrpcpb.RawPutResponse, error) {
	// Your Code Here (1).
	// Hint: Consider using Storage.Modify to store data to be modified
	batch := make([]storage.Modify, 1)
	batch[0].Data = storage.Put{req.Key, req.Value, req.Cf}
	err := server.storage.Write(nil, batch)
	if err != nil {
		return nil, err
	}
	return nil, nil
}

// RawDelete delete the target data from storage and returns the corresponding response
func (server *Server) RawDelete(_ context.Context, req *kvrpcpb.RawDeleteRequest) (*kvrpcpb.RawDeleteResponse, error) {
	// Your Code Here (1).
	// Hint: Consider using Storage.Modify to store data to be deleted
	batch := make([]storage.Modify, 1)
	batch[0].Data = storage.Delete{req.Key, req.Cf}
	err := server.storage.Write(nil, batch)
	if err != nil {
		return nil, err
	}
	return nil, nil
}

// RawScan scan the data starting from the start key up to limit. and return the corresponding result
func (server *Server) RawScan(_ context.Context, req *kvrpcpb.RawScanRequest) (*kvrpcpb.RawScanResponse, error) {
	// Your Code Here (1).
	// Hint: Consider using reader.IterCF
	reader, err := server.storage.Reader(nil)
	if err != nil {
		return nil, err
	}
	defer reader.Close()

	iterator := reader.IterCF(req.Cf)
	defer iterator.Close()

	kvs := make([]*kvrpcpb.KvPair, 0)
	iterator.Seek(req.StartKey)
	for i := 0; i < int(req.Limit) && iterator.Valid(); i++ {
		item := iterator.Item()
		val, _ := item.Value()
		if val == nil {
			kvs = append(kvs, &kvrpcpb.KvPair{Key: item.Key()})
		} else {
			kvs = append(kvs, &kvrpcpb.KvPair{Key: item.Key(), Value: val})
		}
		iterator.Next()
	}
	return &kvrpcpb.RawScanResponse{Kvs: kvs}, nil
}
